// Copyright 2023 Huawei Cloud Computing Technology Co., Ltd.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include "StreamParser.h"
#include <arpa/inet.h>
#include <array>
#include <cassert>
#include <cstring>
#include <linux/prctl.h>
#include <sys/prctl.h>
#include <thread>
#include <unistd.h>

#include "Communication/Socket/VmiSocket.h"
#include "HWSecure/include/securec.h"
#include "Include/msghead.h"
#include "Log/logging.h"

using namespace std;

namespace {
    const int VMI_THREAD_RUNNING = 1;
    const int VMI_THREAD_EXIT = -1;
    const int STREAM_HEAD_SIZE = 8;
    const int STREAM_THREAD_NAME_LEN = 32;
    const int PUC_BUFFER_FACTOR = 16;
    const int STREAM_MAX_BUF_LEN = MAX_MSG_SIZE;
}

/**
 * @brief: constructor
 */
/*lint -save -e1401 */
StreamParse::StreamParse()
{
    m_serviceHandle.fill(nullptr);
}
/*lint -restore */
StreamParse::~StreamParse()
{
}

/**
 * @brief: set handle with type
 * @param [in] type: type
 * @param [in] serviceHandle: handle
 */
void StreamParse::SetServiceHandle(unsigned char type, std::shared_ptr<ServiceHandle> serviceHandle)
{
    m_serviceHandle[type] = serviceHandle;
}

/**
 * @brief: get handle with type
 * @param [in] type: type
 * @param [in] serviceHandle: handle
 */
ServiceHandle* StreamParse::GetServiceHandle(unsigned char type)
{
    return m_serviceHandle[type].get();
}

/**
 * @brief: check packet
 * @param [in] packet: packet
 * @param [in] len: data length
 * @return: packet length or error code
 */
int StreamParse::PacketDelimiter(uint8_t* packet, int len)
{
    if (sizeof(StreamMsgHead) != STREAM_HEAD_SIZE) {
        ERR("recv a error package");
        return -1;
    }
    if (len < static_cast<int>(sizeof(StreamMsgHead))) {
        ERR("invalid data, data must start with StreamMsgHead head");
        return 0;
    }

    StreamMsgHead* messageHead = reinterpret_cast<StreamMsgHead *>(packet);
    if (messageHead->getType() >= VMIMsgType::End || messageHead->getType() <= VMIMsgType::Invalid ||
        messageHead->verifyMsgHead() == false) {
        ERR("Stream msghead verification failed, type:%d, verify:%d", messageHead->getType(), messageHead->verifyMsgHead());
        return -1;
    }

    return messageHead->getPayloadSize();
}

/**
 * @brief: process packet
 * @param [in] packetData: data point
 * @param [out] dataLen: data length
 * @param [in] packetLength: packet length
 * @return: position of data
 */
uint8_t* StreamParse::ProcessMsg(uint8_t* packetData, int& dataLen, int packetLength)
{
    StreamMsgHead* messageHead = reinterpret_cast<StreamMsgHead *>(packetData);
    ServiceHandle* pserviceHandle = GetServiceHandle(messageHead->getType());
    if (pserviceHandle != nullptr) {
        uint8_t* tmp = reinterpret_cast<uint8_t *>(VmiAllocBuffer(packetLength));
        if (tmp != nullptr) {
            int errCode = memcpy_s(tmp, packetLength, packetData, packetLength);
            if (errCode != EOK) {
                // Copy failed, lost this data, but to update the pointer location, because this data is
                // received and stored, but do not process it, pointer moved to the correct location, so
                // as to receive and process the subsequent data.
                ERR("memcpy error");
                packetData += packetLength;
                dataLen -= packetLength;
                VmiFreeBuffer(tmp);
                return packetData;
            }
            VmiBuffer recvBuf(tmp, packetLength);
            pserviceHandle->Handle(recvBuf);
        } else {
            ERR("STREAMPARSE: alloc buf(size %d) failed\n", packetLength);
        }
    } else {
        ERR("STREAMPARSE: msg(type %u) handle is nullptr\n", messageHead->getType()); // Drop
    }
    packetData += packetLength;
    dataLen -= packetLength;
    return packetData;
}

/**
 * @brief: parse socket stream
 * @param: [in] packetDataBuf: packetDataBuf
 * @param: [in] bufLen: bufLen
 * @param: [in] packetData: packetData
 * @param: [out] packetDataLength: packetDataLength
 */
uint8_t* StreamParse::ParseStream(uint8_t* packetDataBuf, int bufLen, uint8_t* packetData, int& packetDataLength)
{
    int packetLength = 0;
    int len = packetDataLength;
    while (len > static_cast<int>(sizeof(StreamMsgHead))) {
        packetLength = PacketDelimiter(packetData, len);
        if (packetLength > 0) {
            packetLength += sizeof(StreamMsgHead);
            if (packetLength <= len) {
                packetData = ProcessMsg(packetData, len, packetLength);
                // Moving the pointer here, if the application memory fails occasionally, a message packet will be
                // discarded, and if the application is not persistent, the loop will not be dead here.
                continue;
            } else if (packetLength < bufLen) {
                // This is the normal branch, but the current temporary data received
                // is not enough for a complete message, return to continue receiving
                break;
            } else { 
                // It is impossible to receive message processing longer than BUF. It may be a packet header
                // delimitation error, byte-by-byte parsing, and retrieving the packet header.
                ERR("STREAMPARSE: parse failed. PacketLen(%d)\n", packetLength);
            }
        } else if (packetLength == 0) {
            // The return length of the analytic function is equal to 0, indicating that the current received
            // data is insufficient for the parse length. Continue
            break;
        } else {
            // The return length of the analytic function is less than 0, which means that the parsing fails. It
            // may be a packet header delimitation error, byte-by-byte parsing and rediscovering the packet header.
            ERR("STREAMPARSE: parse failed\n");
        }

        // Byte-by-byte parsing to retrieve the header
        packetData += 1;
        len -= 1;
    }

    if ((packetData - packetDataBuf + len) > (bufLen - (STREAM_MAX_BUF_LEN / PUC_BUFFER_FACTOR))) {
        int errorCode = memmove_s(packetDataBuf, bufLen, packetData, len);
        if (errorCode != EOK) {
            // There is no need to terminate the process, the copy fails, and the packetData pointer moves 
            // backwards to ensure that the packetRecvBufferFree pointer moves to the correct location.
            ERR("memmove_s error");
        }
        packetData = packetDataBuf;
    }
    packetDataLength = len;
    return packetData;
}

/**
 * @brief: process recv message
 * @param [in] recvBuf: recv buff
 * @param [in] streamParse: streamParse
 * @param [out] dataLen：buff length
 */
void StreamParseThread::ThreadPara::ProcessRecvMsg(uint8_t* recvBuf, StreamParse& streamParse, int& dataLen)//lint !e1072
{
    uint8_t* packetRecvBuffer = recvBuf;
    int byte = static_cast<int>((uintptr_t)packetRecvBuffer & 0x7);
    uint8_t* packetDataBuf = packetRecvBuffer;
    int bufLen = STREAM_MAX_BUF_LEN;
    int freeLen = bufLen; // free length in recv buffer
    uint8_t* packetRecvBufferFree = packetDataBuf;
    uint8_t* packetData = packetRecvBufferFree;
    int offset = 0;

    while (m_threadStatus == VMI_THREAD_RUNNING) {
        VmiBuffer recvBuf(packetRecvBufferFree, freeLen);
        byte = m_vmiSocket->Recv(recvBuf);
        if ((byte > 0) && (byte <= freeLen)) {
            freeLen -= byte;
            dataLen += byte;
            packetData = streamParse.ParseStream(packetDataBuf, bufLen, packetData, dataLen);
            packetRecvBufferFree = packetData + dataLen;
            offset = packetRecvBufferFree - packetDataBuf;
            freeLen = bufLen - offset;
        } else {
            LOG_RATE_LIMIT(ANDROID_LOG_ERROR, 1, "socket(%p) recv fail, ret = %d\n", m_vmiSocket.get(), byte);
            if (byte == SOCKET_RECV_FAIL_DISCONN) {
                INFO("SOCKET_RECV_FAIL_DISCONN\n");
                break;
            }
        }
    }
}

/**
 * @brief: thread process
 * @param [in] streamParseThreadObj: thread parameter
 */
void Taskentry(std::shared_ptr<StreamParseThread::ThreadPara> streamParseThreadObj)
{
    StreamParse* streamParse = streamParseThreadObj->m_streamParse;
    uint8_t* packetRecvBuffer = nullptr;
    int dataLen = 0;  // data length in recv buffer
    INFO("STREAMPARSE: thread entering");

    char threadName[STREAM_THREAD_NAME_LEN] = {'\0'};
    int code = snprintf_s(threadName, STREAM_THREAD_NAME_LEN, sizeof(threadName), "SP-%p",
        streamParseThreadObj->m_vmiSocket.get());
    if (code < 0) {
        ERR("write threadName failed");
    }

    // print like 0xHHHHHHHHHHHHHHHH
    prctl(PR_SET_NAME, threadName);

    packetRecvBuffer = reinterpret_cast<uint8_t *>(malloc(STREAM_MAX_BUF_LEN));
    if ((packetRecvBuffer != nullptr) && (streamParse != nullptr)) {
        INFO("STREAMPARSE: thread running");

        streamParseThreadObj->ProcessRecvMsg(packetRecvBuffer, *streamParse, dataLen);
        ERR("STREAMPARSE: thread exit, drop data (size %d)\n", dataLen);
        free(packetRecvBuffer);
        packetRecvBuffer = nullptr;
    } else {
        ERR("Stream recv task failed! alloc recvbuf(%p,size %d) streamParse(%p)", packetRecvBuffer, STREAM_MAX_BUF_LEN,
            streamParse);
        if (packetRecvBuffer != nullptr) {
            free(packetRecvBuffer);
            packetRecvBuffer = nullptr;
        }
    }

    INFO("STREAMPARSE: thread exited");
    return;
}

/**
 * @brief: stream parse thread
 * @param [in] vmiSocket: socket
 */
StreamParseThread::StreamParseThread(std::shared_ptr<VmiSocket> vmiSocket)
{
    m_vmiSocket = vmiSocket;
    m_threadParameter = nullptr;
    m_task = nullptr;
}

/**
 * @brief: deconstructor
 */
StreamParseThread::~StreamParseThread()
{
    INFO("STREAMPARSE: destruct\n");
    if (Stop() != 0){
        ERR("stop failed");
    }
    m_vmiSocket = nullptr;
    m_threadParameter = nullptr;
    m_task = nullptr;
    m_streamParse = nullptr;
}

/**
 * @brief: set m_streamParse
 */
void StreamParseThread::SetStreamParse(StreamParse* parse)
{
    m_streamParse = parse;
}

/**
 * @brief: start thread
 */
int StreamParseThread::Start()
{
    lock_guard<mutex> lck(m_lock);
    if (m_threadParameter == nullptr) {
        m_threadParameter = std::shared_ptr<ThreadPara>(new (std::nothrow) StreamParseThread::ThreadPara);
        if (m_threadParameter == nullptr) {
            ERR("NO MEM");
            return -1;
        }

        m_threadParameter->m_vmiSocket = m_vmiSocket;
        m_threadParameter->m_threadStatus = VMI_THREAD_RUNNING;
        m_threadParameter->m_streamParse = m_streamParse;
        m_task = new (std::nothrow) thread(Taskentry, m_threadParameter);
        if (m_task == nullptr) {
            ERR("NO MEM");
            return -1;
        }
    }

    return 0;
}

/**
 * @brief: stop thread, caller wait it exit
 */
int StreamParseThread::Stop()
{
    lock_guard<mutex> lck(m_lock);
    if (m_threadParameter != nullptr) {
        INFO("STREAMPARSE: set thread to exit\n");
        m_threadParameter->m_threadStatus = VMI_THREAD_EXIT;
        if (m_task != nullptr) {
            m_task->join();
            delete m_task;
            m_task = nullptr;
        }
        m_vmiSocket->CloseSocket();  
        m_threadParameter->m_streamParse = nullptr;
    } else {
        INFO("STREAMPARSE: thread is not running");
    }

    return 0;
}
